package com.upcoming.data.storm.fileRead;

import com.upcoming.data.storm.util.MyUtil;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-23 9:57
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class FileSpout implements IRichSpout {

    private static final Logger logger = LoggerFactory.getLogger(FileSpout.class);

    private SpoutOutputCollector collector;

    private String fileName;

    private BufferedReader reader;

    private AtomicLong linesRead;


    public FileSpout() {
        MyUtil.outLog2NC(this, "new FileSpout()");
        logger.info("new FileSpout()");
    }



    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("data"));
    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.collector = spoutOutputCollector;

        linesRead = new AtomicLong(0);

        try {
            fileName = (String) map.get("myfile");

            reader = new BufferedReader(new FileReader(fileName));
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void close() {
        try {
            reader.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void activate() {

    }

    public void deactivate() {

    }

    public void nextTuple() {
//        MyUtil.outLog2NC(this, "nextTuple()");
        try {
            String line = reader.readLine();
            logger.info(line);
            if (line != null) {
                MyUtil.outLog2NC(this, line);
                long id = linesRead.incrementAndGet();
                this.collector.emit(new Values(line), id);
            }
            Thread.sleep(200);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void ack(Object o) {
        MyUtil.outLog2NC(this, "ack:" + o);
    }

    public void fail(Object o) {
        MyUtil.outLog2NC(this, "fail:" + o);
    }
}
